package com.atguigu.gmall.rabbit.config;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.rabbit.model.GmallCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.awt.*;
import java.util.concurrent.TimeUnit;

/**
 * MQProducerConfig类产生对象：既是ConfirmCallback接口类型 又是ReturnCallback接口类型
 *
 * @author: atguigu
 * @create: 2023-09-18 09:15
 */
@Slf4j
@Component
public class MQProducerConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplate redisTemplate;


    /**
     * 项目启动成功后（RabbitTemplate对象在IOC容器中）自动执行一次
     */
    @PostConstruct
    public void init() {
        //设置交换机确认回调
        rabbitTemplate.setConfirmCallback(this);
        //设置队列确认异常回调
        rabbitTemplate.setReturnCallback(this);
        log.info("[RabbitMQ]生产者，项目启动后执行生产者确认设置");

    }


    /**
     * 生产者确认回调-交换机回调（正常，异常）
     *
     * @param correlationData 相关数据,只有当调用模板对象发送业务消息同时设置相关数据
     * @param ack             true：正常, false：异常
     * @param cause           异常原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            log.error("交换机异常回调，原因：{},进入到重试逻辑。", cause);
            if (correlationData != null && StringUtils.isNotBlank(correlationData.getId())) {
                this.retryMessage(correlationData.getId());
            }
        }
    }


    /**
     * 生产者确认回调-队列确认异常。只要交换机路由消息到队列失败就会触发该方法回调
     * 特殊情况：延迟插件必然会触发异常！！
     *
     * @param message    MQ消息对象.
     * @param replyCode  异常码.
     * @param replyText  异常消息.
     * @param exchange   交换机名称.
     * @param routingKey 路由键名称.
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("[生产者-队列异常确认],消息路由queue失败，应答码={}，原因={}，交换机={}，路由键={}，MQ消息={},进入到重试逻辑.", replyCode, replyText, exchange, routingKey, message.toString());
        //1.从MQ消息对象中属性信息中获取头信息 spring_returned_message_correlation -> 25fc60a2aa2d4981a6aaa140a0be5ec6
        String id = message.getMessageProperties().getHeader("spring_returned_message_correlation");
        if (StringUtils.isNotBlank(id)) {
            //2.进行重试
            this.retryMessage(id);
        }
    }


    /**
     * 当生产者发送消息异常，在人工介入前程序自动进行重试
     *
     * @param id 相关数据标识，根据该标识获取存在Redis中相关数据信息
     */
    private void retryMessage(String id) {
        //1.构建相关数据Key
        String key = "mq:" + id;

        //2.查询Redis中存放相关数据-注意该数据存采用Strin    g
        String correlationDataStr = (String) redisTemplate.opsForValue().get(key);
        GmallCorrelationData gmallCorrelationData = JSON.parseObject(correlationDataStr, GmallCorrelationData.class);

        //重发对于延迟消息处理方式一：忽略延迟消息
        if (gmallCorrelationData.isDelay()) {
            return;
        }
        //3.判断重试是否超限 阈值：3次
        int retryCount = gmallCorrelationData.getRetryCount();
        if (retryCount > 2) {
            //4.如果重试超限，写入生产者消息异常表（将来由人工处理）
            log.error("[RabbitMQ]消息重发超限：相关数据：{}", correlationDataStr);
            return;
        }
        //5.如果重试未超限，再次发送业务消息+相关数据  更新重试次数
        //5.1 更新重试次数
        gmallCorrelationData.setRetryCount(retryCount + 1);
        log.info("正在执行第：{}次重发消息。", retryCount + 1);
        Long ttl = redisTemplate.getExpire(key);
        redisTemplate.opsForValue().set(key, JSON.toJSONString(gmallCorrelationData), ttl, TimeUnit.SECONDS);

        //5.2 将发送失败消息再次进行重发
        if (gmallCorrelationData.isDelay()) {
            //延迟消息重发
            rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(), gmallCorrelationData.getRoutingKey(), gmallCorrelationData.getMessage(), message -> {
                message.getMessageProperties().setDelay(gmallCorrelationData.getDelayTime() * 1000);
                return message;
            }, gmallCorrelationData);
        } else {
            //普通消息重发
            rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(), gmallCorrelationData.getRoutingKey(), gmallCorrelationData.getMessage(), gmallCorrelationData);
        }
    }
}
